﻿using RockBus.Configuration.Interfaces;
using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.Routing.Interfaces;
using RockBus.Threading;
using RockBus.Utilities;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Reflection;

namespace RockBus.Routing
{
    public class CorrelationManager : ICorrelationManager
    {
        public CorrelationManager(
            ILogger logger,
            IQueueManager queueManager,
            IInternalConfiguration configuration)
        {
            this.Logger = logger;
            this.QueueManager = queueManager;
            this.Configuration = configuration;

            this.InboundCorrelationDictionary = new ConcurrentDictionary<Guid, TimeoutHelper<IMessageContext>>();
            this.OutboundCorrelationDictionary = new ConcurrentDictionary<Guid, TimeoutHelper<IMessageContext>>();
        }

        private ILogger Logger { get; set; }

        private IQueueManager QueueManager { get; set; }

        private IInternalConfiguration Configuration { get; set; }

        private readonly ConcurrentDictionary<Guid, TimeoutHelper<IMessageContext>> InboundCorrelationDictionary;

        private readonly ConcurrentDictionary<Guid, TimeoutHelper<IMessageContext>> OutboundCorrelationDictionary;

        public bool TryStartInboundCorrelation(IMessageContext messageContext)
        {
            // The EvaluationManager creates MessageContexts without any RoutingHeaders when it is supposed to be for correlation purposes.
            //  Therefore, if the MessageContext contains a RoutingHeader, it's not intended for an inbound correlation
            if (!messageContext.RoutingHeaders.IsNullOrEmpty())
            {
                return false;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "CorrelationManager - Starting inbound correlation for {0}", messageContext);
            var th = new TimeoutHelper<IMessageContext>((s, t) =>
            {
                this.TryRemoveAfterTimeout(s, t, this.InboundCorrelationDictionary);
            }, messageContext, this.Configuration.CorrelationTimeoutMs);
            this.InboundCorrelationDictionary.AddOrUpdate(messageContext.ProcessingData.CorrelationId, th, (id, tp) => { return th; });
            return true;
        }

        public bool TryStartOutboundCorrelation(IMessageContext messageContext)
        {
            if (!messageContext.IsCorrelatedSubscriberEndpoint())
            {
                return false;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "CorrelationManager - Starting outbound correlation for {0}", messageContext);
            var th = new TimeoutHelper<IMessageContext>((s, t) =>
            {
                this.TryRemoveAfterTimeout(s, t, this.OutboundCorrelationDictionary);
            }, messageContext, this.Configuration.CorrelationTimeoutMs);
            this.OutboundCorrelationDictionary.AddOrUpdate(messageContext.ProcessingData.CorrelationId, th, (id, tp) => { return th; });
            return true;
        }

        public bool TrySatisfyInboundCorrelation(IMessageContext messageContext, out IMessageContext correlatedMessageContext)
        {
            correlatedMessageContext = null;

            TimeoutHelper<IMessageContext> th = null;
            if (!this.InboundCorrelationDictionary.TryRemove(messageContext.ProcessingData.CorrelationId, out th))
            {
                this.Logger.LogWarning(MethodBase.GetCurrentMethod(), "Unknown CorrelationId for {0}", messageContext);
                return false;
            }

            th.Complete();

            correlatedMessageContext = th.State;
            if (th.IsTimedOut)
            {
                correlatedMessageContext.Exception = new TimeoutException("No response was produced within the specified timeout.");
                return true;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Satisfying inbound correlation for {0}", correlatedMessageContext);
            correlatedMessageContext.SetRoutingHeader(messageContext.RoutingHeaders.FirstOrDefault());
            correlatedMessageContext.ResponseMessage = messageContext.ResponseMessage;
            correlatedMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundResponse;
            return true;
        }

        public bool TrySatisfyOutboundCorrelation(IMessageContext messageContext, out IMessageContext correlatedMessageContext)
        {
            correlatedMessageContext = null;

            TimeoutHelper<IMessageContext> th = null;
            if (!this.OutboundCorrelationDictionary.TryRemove(messageContext.ProcessingData.CorrelationId, out th))
            {
                this.Logger.LogWarning(MethodBase.GetCurrentMethod(), "Unknown CorrelationId for {0}", messageContext);
                return false;
            }

            th.Complete();

            correlatedMessageContext = th.State;
            if (th.IsTimedOut)
            {
                correlatedMessageContext.Exception = new TimeoutException("No response was produced within the specified timeout.");
                return false;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Satisfying outbound correlation for {0}", correlatedMessageContext);
            correlatedMessageContext.ResponseMessage = messageContext.ResponseMessage;
            correlatedMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundResponse;
            return true;
        }

        private void TryRemoveAfterTimeout(IMessageContext messageContext, bool isTimedOut, ConcurrentDictionary<Guid, TimeoutHelper<IMessageContext>> dict)
        {
            // If the TimerHelper didn't timeout, the entry was removed from the dictionary by a satisfied response
            //  Checking for IsTimedOut will prevent superfluous checking if the entry is still present in the Dictionary
            if (!isTimedOut)
            {
                return;
            }

            TimeoutHelper<IMessageContext> th = null;
            if (dict.TryRemove(messageContext.ProcessingData.CorrelationId, out th))
            {
                if (!th.IsTimedOut)
                {
                    // This should not be possible, log if it still did occur
                    this.Logger.LogWarning(MethodBase.GetCurrentMethod(), "Impossible: Dictionary entry was removed after timeout, but TimeoutHelper.IsTimedOut was false");
                }

                var mc = th.State;
                mc.Exception = new TimeoutException("No response was produced within the specified timeout.");
                this.QueueManager.ProcessException(mc);
            }
        }
    }
}